package im.engure.common.client.netty;

import im.engure.common.client.netty.handler.NettyClientHandler;
import im.engure.common.exception.RPCException;
import im.engure.common.protocol.Invocation;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NettyClient {

    public NettyClientHandler nettyClientHandler = null;

    private final ExecutorService executorService = Executors.newCachedThreadPool();

    public void start(String hostName, Integer port) {
        try {
            nettyClientHandler = new NettyClientHandler();
            NioEventLoopGroup group = new NioEventLoopGroup();

            Bootstrap b = new Bootstrap();
            ChannelFuture channelFuture = b.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                            ChannelPipeline pipeline = nioSocketChannel.pipeline();
                            pipeline.addLast("decoder", new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                            pipeline.addLast("encoder", new ObjectEncoder());
                            pipeline.addLast("handler", nettyClientHandler);
                        }
                    })
                    .connect(hostName, port);

            channelFuture.sync();
            Channel ch = channelFuture.channel();

            //连接关闭后自动关闭线程池、事件循环组
            ch.closeFuture().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    try {
                        group.shutdownGracefully();
                    } finally {
                        executorService.shutdown();
                    }
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public String send(String hostName, Integer port, Invocation invocation) {
        if (nettyClientHandler == null) {
            start(hostName, port);
        }

        nettyClientHandler.setInvocation(invocation);

        try {
            //交给 executorServices 提交处理
            return (String) executorService.submit(nettyClientHandler).get();
        } catch (InterruptedException | ExecutionException e) {
            //e.printStackTrace();
            RPCException ex = new RPCException();
            ex.addSuppressed(e);
            throw ex;
        }
        //return null;
    }
}